package drds.binlog.store;

import drds.binlog.common.Authors;
import drds.binlog.common.StorageBatchMode;
import drds.binlog.common.position.LogPosition;
import drds.binlog.common.position.Position;
import drds.binlog.common.position.PositionRange;
import drds.binlog.data_object.EntryType;
import drds.binlog.data_object.EventType;
import drds.common.$;
import drds.common.Author;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 基于内存buffer构建内存memory store
 * <p>
 * <pre>
 * 变更记录：
 * 1. 新增BatchMode类型，支持按内存大小获取批次数据，内存大小更加可控.
 *   a. put操作，会首先根据bufferSize进行控制，然后再进行bufferSize * bufferMemUnit进行控制. 因存储的内容是以Event，如果纯依赖于memsize进行控制，会导致RingBuffer出现动态伸缩
 * </pre>
 */
@Author(name = Authors.LI_YANG)
public class EventStore extends AbstractCleanUp implements IEventStore<Event>, CleanUp
{

    private static final long initActionIndex = -1;
    private int eventsSize = 16 * 1024;
    private int memorySizePerEvent = 1024;                         // memsize的单位，默认为1kb大小
    private int lastEventsIndex;
    private Event[] events;

    // 记录下put/get/ack操作的三个下标
    /**
     * 初始值是-1
     */
    private AtomicLong putActionIndex = new AtomicLong(initActionIndex); // 代表当前put操作最后一次写操作发生的位置
    /**
     * 初始值是-1
     */
    private AtomicLong getActionIndex = new AtomicLong(initActionIndex); // 代表当前get操作读取的最后一条的位置
    /**
     * 初始值是-1
     */
    private AtomicLong ackActionIndex = new AtomicLong(initActionIndex); // 代表当前ack操作的最后一条的位置


    // 记录下put/get/ack操作的三个memsize大小
    private AtomicLong putActionMemorySize = new AtomicLong(0);
    private AtomicLong getActionMemorySize = new AtomicLong(0);
    private AtomicLong ackActionMemorySize = new AtomicLong(0);

    // 记录下put/get/ack操作的三个execTime
    private AtomicLong putActionDateTimeLong = new AtomicLong(System.currentTimeMillis());
    private AtomicLong getActionDateTimeLong = new AtomicLong(System.currentTimeMillis());
    private AtomicLong ackActionDateTimeLong = new AtomicLong(System.currentTimeMillis());

    // 记录下put/get/ack操作的三个table rows
    private AtomicLong putActionTableRowNumber = new AtomicLong(0);
    private AtomicLong getActionTableRowNumber = new AtomicLong(0);
    private AtomicLong ackActionTableRowNumber = new AtomicLong(0);

    // 阻塞put/get操作控制信号
    private ReentrantLock lock = new ReentrantLock();
    private Condition notFull = lock.newCondition();//put
    private Condition notEmpty = lock.newCondition();//get

    private StorageBatchMode storageBatchMode = StorageBatchMode.item_size;           // 默认为内存大小模式
    private boolean ddlIsolation = false;

    public EventStore()
    {

    }

    public EventStore(StorageBatchMode storageBatchMode)
    {
        this.storageBatchMode = storageBatchMode;
    }

    public void start() throws StoreException
    {
        super.start();
        if (Integer.bitCount(eventsSize) != 1)
        {
            throw new IllegalArgumentException("eventsSize must be a power of 2");
        }

        lastEventsIndex = eventsSize - 1;
        events = new Event[eventsSize];
    }

    public void stop() throws StoreException
    {
        super.stop();

        cleanUp();
    }

    //
    public void put(List<Event> eventList) throws InterruptedException, StoreException
    {
        if (!$.isHasData(eventList))
        {
            return;
        }

        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try
        {
            try
            {
                while (!checkUnPutSlotAt(putActionIndex.get() + eventList.size()))
                { // 检查是否有空位
                    notFull.await(); // wait until not full
                }
            } catch (InterruptedException ie)
            {
                notFull.signal(); // propagate to non-interrupted thread
                throw ie;
            }
            addIntoEvents(eventList);
            if (Thread.interrupted())
            {
                throw new InterruptedException();
            }
        } finally
        {
            lock.unlock();
        }
    }

    public boolean put(List<Event> eventList, long timeout, TimeUnit timeUnit) throws InterruptedException, StoreException
    {
        if (eventList == null || eventList.isEmpty())
        {
            return true;
        }

        long nanos = timeUnit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try
        {
            for (; ; )
            {
                if (checkUnPutSlotAt(putActionIndex.get() + eventList.size()))
                {
                    addIntoEvents(eventList);
                    return true;
                }
                if (nanos <= 0)
                {
                    return false;
                }

                try
                {
                    nanos = notFull.awaitNanos(nanos);
                } catch (InterruptedException ie)
                {
                    notFull.signal(); // propagate to non-interrupted thread
                    throw ie;
                }
            }
        } finally
        {
            lock.unlock();
        }
    }

    public boolean tryPut(List<Event> eventList) throws StoreException
    {
        if (eventList == null || eventList.isEmpty())
        {
            return true;
        }

        final ReentrantLock lock = this.lock;
        lock.lock();
        try
        {
            if (!checkUnPutSlotAt(putActionIndex.get() + eventList.size()))
            {
                return false;
            } else
            {
                addIntoEvents(eventList);
                return true;
            }
        } finally
        {
            lock.unlock();
        }
    }

    public void put(Event event) throws InterruptedException, StoreException
    {
        put(Arrays.asList(event));
    }

    public boolean put(Event event, long timeout, TimeUnit unit) throws InterruptedException, StoreException
    {
        return put(Arrays.asList(event), timeout, unit);
    }

    public boolean tryPut(Event event) throws StoreException
    {
        return tryPut(Arrays.asList(event));
    }

    /**
     * 执行具体的put操作
     */
    private void addIntoEvents(List<Event> eventList)
    {
        long startIndex = putActionIndex.get();
        long endIndex = startIndex + eventList.size();

        // 先写数据，再更新对应的cursor,并发度高的情况，putSequence会被get请求可见，拿出了ringbuffer中的老的Entry值
        for (long nextIndex = startIndex + 1; nextIndex <= endIndex; nextIndex++)
        {
            events[getEventIndex(nextIndex)] = eventList.get((int) (nextIndex - startIndex - 1));
        }

        putActionIndex.set(endIndex);

        // 记录一下gets memsize信息，方便快速检索
        if (storageBatchMode.isMemorySize())
        {
            long size = 0;
            for (Event event : eventList)
            {
                size += getEventSize(event);
            }

            putActionMemorySize.getAndAdd(size);
        }
        profiling(eventList, Action.put);
        // tell other threads that store is not empty
        notEmpty.signal();
    }

    public Events get(Position start, int batchSize) throws InterruptedException, StoreException
    {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try
        {
            try
            {
                while (!checkUnGetSlotAt((LogPosition) start, batchSize))
                {

                    notEmpty.await();
                }

            } catch (InterruptedException ie)
            {
                notEmpty.signal(); // propagate to non-interrupted thread
                throw ie;
            }

            return doGet(start, batchSize);
        } finally
        {
            lock.unlock();
        }
    }

    public Events get(Position start, int batchSize, long timeout, TimeUnit timeUnit) throws InterruptedException,
            StoreException
    {
        long nanos = timeUnit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try
        {
            for (; ; )
            {
                if (checkUnGetSlotAt((LogPosition) start, batchSize))
                {
                    return doGet(start, batchSize);
                }

                if (nanos <= 0)
                {
                    // 如果时间到了，有多少取多少
                    return doGet(start, batchSize);
                }

                try
                {
                    nanos = notEmpty.awaitNanos(nanos);
                } catch (InterruptedException ie)
                {
                    notEmpty.signal(); // propagate to non-interrupted thread
                    throw ie;
                }

            }
        } finally
        {
            lock.unlock();
        }
    }

    public Events tryGet(Position start, int batchSize) throws StoreException
    {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try
        {
            return doGet(start, batchSize);
        } finally
        {
            lock.unlock();
        }
    }

    private Events doGet(Position start, int batchSize) throws StoreException
    {
        LogPosition startPosition = (LogPosition) start;

        long current = getActionIndex.get();
        long maxAbleSequence = putActionIndex.get();
        long next = current;
        long end = current;
        // 如果startPosition为null，说明是第一次，默认+1处理
        if (startPosition == null || !startPosition.getEntryPosition().isIncluded())
        { // 第一次订阅之后，需要包含一下start位置，防止丢失第一条记录
            next = next + 1;
        }

        if (current >= maxAbleSequence)
        {
            return new Events();
        }

        Events events = new Events();
        List<Event> eventList = events.getEventList();
        long memsize = 0;
        if (storageBatchMode.isItemSize())
        {
            end = (next + batchSize - 1) < maxAbleSequence ? (next + batchSize - 1) : maxAbleSequence;
            // 提取数据并返回
            for (; next <= end; next++)
            {
                Event event = this.events[getEventIndex(next)];
                if (ddlIsolation && isDdl(event.getEventType()))
                {
                    // 如果是ddl隔离，直接返回
                    if (eventList.size() == 0)
                    {
                        eventList.add(event);// 如果没有DML事件，加入当前的DDL事件
                        end = next; // 更新end为当前
                    } else
                    {
                        // 如果之前已经有DML事件，直接返回了，因为不包含当前next这记录，需要回退一个位置
                        end = next - 1; // next-1一定大于current，不需要判断
                    }
                    break;
                } else
                {
                    eventList.add(event);
                }
            }
        } else
        {
            long maxMemSize = batchSize * memorySizePerEvent;
            for (; memsize <= maxMemSize && next <= maxAbleSequence; next++)
            {
                // 永远保证可以取出第一条的记录，避免死锁
                Event event = this.events[getEventIndex(next)];
                if (ddlIsolation && isDdl(event.getEventType()))
                {
                    // 如果是ddl隔离，直接返回
                    if (eventList.size() == 0)
                    {
                        eventList.add(event);// 如果没有DML事件，加入当前的DDL事件
                        end = next; // 更新end为当前
                    } else
                    {
                        // 如果之前已经有DML事件，直接返回了，因为不包含当前next这记录，需要回退一个位置
                        end = next - 1; // next-1一定大于current，不需要判断
                    }
                    break;
                } else
                {
                    eventList.add(event);
                    memsize += getEventSize(event);
                    end = next;// 记录end位点
                }
            }

        }

        PositionRange<LogPosition> positionRange = new PositionRange<LogPosition>();
        events.setPositionRange(positionRange);

        positionRange.setStartPosition(LogPositions.createPosition(eventList.get(0)));
        positionRange.setEndPosition(LogPositions.createPosition(eventList.get(events.getEventList().size() - 1)));
        // 记录一下是否存在可以被ack的点

        for (int i = eventList.size() - 1; i >= 0; i--)
        {
            Event event = eventList.get(i);
            if (EntryType.transaction_begin == event.getEntryType()
                    || EntryType.transaction_end == event.getEntryType() || isDdl(event.getEventType()))
            {
                // 将事务头/尾设置可被为ack的点
                positionRange.setAckPosition(LogPositions.createPosition(event));
                break;
            }
        }

        if (getActionIndex.compareAndSet(current, end))
        {
            getActionMemorySize.addAndGet(memsize);
            notFull.signal();
            profiling(events.getEventList(), Action.get);
            return events;
        } else
        {
            return new Events();
        }
    }

    public LogPosition getFirstPosition() throws StoreException
    {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try
        {
            long firstSeqeuence = ackActionIndex.get();
            if (firstSeqeuence == initActionIndex && firstSeqeuence < putActionIndex.get())
            {
                // 没有ack过数据
                Event event = events[getEventIndex(firstSeqeuence + 1)]; // 最后一次ack为-1，需要移动到下一条,included
                // = false
                return LogPositions.createPosition(event, false);
            } else if (firstSeqeuence > initActionIndex && firstSeqeuence < putActionIndex.get())
            {
                // ack未追上put操作
                Event event = events[getEventIndex(firstSeqeuence + 1)]; // 最后一次ack的位置数据
                // + 1
                return LogPositions.createPosition(event, true);
            } else if (firstSeqeuence > initActionIndex && firstSeqeuence == putActionIndex.get())
            {
                // 已经追上，store中没有数据
                Event event = events[getEventIndex(firstSeqeuence)]; // 最后一次ack的位置数据，和last为同一条，included
                // = false
                return LogPositions.createPosition(event, false);
            } else
            {
                // 没有任何数据
                return null;
            }
        } finally
        {
            lock.unlock();
        }
    }

    public LogPosition getLatestPosition() throws StoreException
    {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try
        {
            long latestSequence = putActionIndex.get();
            if (latestSequence > initActionIndex && latestSequence != ackActionIndex.get())
            {
                Event event = events[(int) putActionIndex.get() & lastEventsIndex]; // 最后一次写入的数据，最后一条未消费的数据
                return LogPositions.createPosition(event, true);
            } else if (latestSequence > initActionIndex && latestSequence == ackActionIndex.get())
            {
                // ack已经追上了put操作
                Event event = events[(int) putActionIndex.get() & lastEventsIndex]; // 最后一次写入的数据，included
                // =
                // false
                return LogPositions.createPosition(event, false);
            } else
            {
                // 没有任何数据
                return null;
            }
        } finally
        {
            lock.unlock();
        }
    }

    public void ack(Position position) throws StoreException
    {
        cleanUpUntil(position);
    }

    public void cleanUpUntil(Position position) throws StoreException
    {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try
        {
            long sequence = ackActionIndex.get();
            long maxSequence = getActionIndex.get();

            boolean hasMatch = false;
            long memsize = 0;
            // ack没有list，但有已存在的foreach，还是节省一下list的开销
            long localExecTime = 0L;
            int deltaRows = 0;
            for (long next = sequence + 1; next <= maxSequence; next++)
            {
                Event event = events[getEventIndex(next)];
                if (localExecTime == 0 && event.getExecuteDateTimeLong() > 0)
                {
                    localExecTime = event.getExecuteDateTimeLong();
                }
                deltaRows += event.getRowsCount();
                memsize += getEventSize(event);
                boolean match = LogPositions.checkPosition(event, (LogPosition) position);
                if (match)
                {// 找到对应的position，更新ack seq
                    hasMatch = true;

                    if (storageBatchMode.isMemorySize())
                    {
                        ackActionMemorySize.addAndGet(memsize);
                        // 尝试清空buffer中的内存，将ack之前的内存全部释放掉
                        for (long index = sequence + 1; index < next; index++)
                        {
                            events[getEventIndex(index)] = null;// 设置为null
                        }
                    }

                    if (ackActionIndex.compareAndSet(sequence, next))
                    {// 避免并发ack
                        notFull.signal();
                        ackActionTableRowNumber.addAndGet(deltaRows);
                        if (localExecTime > 0)
                        {
                            ackActionDateTimeLong.lazySet(localExecTime);
                        }
                        return;
                    }
                }
            }
            if (!hasMatch)
            {// 找不到对应需要ack的position
                throw new StoreException("no match ack position" + position.toString());
            }
        } finally
        {
            lock.unlock();
        }
    }

    public void rollback() throws StoreException
    {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try
        {
            getActionIndex.set(ackActionIndex.get());
            getActionMemorySize.set(ackActionMemorySize.get());
        } finally
        {
            lock.unlock();
        }
    }

    public void cleanUp() throws StoreException
    {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try
        {
            putActionIndex.set(initActionIndex);
            getActionIndex.set(initActionIndex);
            ackActionIndex.set(initActionIndex);

            putActionMemorySize.set(0);
            getActionMemorySize.set(0);
            ackActionMemorySize.set(0);
            events = null;
        } finally
        {
            lock.unlock();
        }
    }


    private long getMinActionIndexInGetAndAckActionIndex()
    {
        long get = getActionIndex.get();
        long ack = ackActionIndex.get();
        return ack <= get ? ack : get;
    }

    /**
     * 查询是否有空位
     */
    private boolean checkUnPutSlotAt(final long putActionIndex)
    {
        final long $ = putActionIndex - eventsSize;
        final long minActionIndexInGetAndAckActionIndex = getMinActionIndexInGetAndAckActionIndex();
        if ($ > minActionIndexInGetAndAckActionIndex)
        { // 刚好追上一轮
            return false;
        } else
        {
            // 在bufferSize模式上，再增加memSize控制
            if (storageBatchMode.isMemorySize())
            {
                final long $memorySize = putActionMemorySize.get() - ackActionMemorySize.get();
                if ($memorySize < eventsSize * memorySizePerEvent)
                {
                    return true;
                } else
                {
                    return false;
                }
            } else
            {
                return true;
            }
        }
    }

    /**
     * 检查是否存在需要get的数据,并且数量>=canalBatchSize
     */
    private boolean checkUnGetSlotAt(LogPosition logPosition, int batchSize)
    {
        if (storageBatchMode.isItemSize())
        {
            //初始值是-1
            long getActionIndex = this.getActionIndex.get();
            long putActionIndex = this.putActionIndex.get();
            //
            long index = getActionIndex;
            if (logPosition == null || !logPosition.getEntryPosition().isIncluded())
            { // 第一次订阅之后，需要包含一下start位置，防止丢失第一条记录
                index = index + 1;// 少一条数据
            }

            if (getActionIndex < putActionIndex && index - 1 + batchSize <= putActionIndex)//这里需要减少一条
            {
                return true;
            } else
            {
                return false;
            }
        } else
        {
            // 处理内存大小判断
            long getActionMemorySize = this.getActionMemorySize.get();
            long putActionMemorySize = this.putActionMemorySize.get();

            if (putActionMemorySize - getActionMemorySize >= batchSize * memorySizePerEvent)
            {
                return true;
            } else
            {
                return false;
            }
        }
    }

    private long getEventSize(Event event)
    {
        // 直接返回binlog中的事件大小
        return event.getRawLength();
    }

    private int getEventIndex(long eventIndex)
    {
        return (int) eventIndex & lastEventsIndex;
    }

    private boolean isDdl(EventType eventType)
    {
        return eventType == EventType.alter || eventType == EventType.create || eventType == EventType.erase
                || eventType == EventType.rename || eventType == EventType.truncate || eventType == EventType.create_index
                || eventType == EventType.drop_index;
    }

    private void profiling(List<Event> eventList, Action action)
    {
        long executeTime = 0L;
        int rowsCount = 0;
        if (eventList != null && !eventList.isEmpty())
        {
            for (Event event : eventList)
            {
                if (executeTime == 0 && event.getExecuteDateTimeLong() > 0)
                {
                    executeTime = event.getExecuteDateTimeLong();
                }
                rowsCount += event.getRowsCount();
            }
        }
        switch (action)
        {
            case put:
                putActionTableRowNumber.addAndGet(rowsCount);
                if (executeTime > 0)
                {
                    putActionDateTimeLong.lazySet(executeTime);
                }
                break;
            case get:
                getActionTableRowNumber.addAndGet(rowsCount);
                if (executeTime > 0)
                {
                    getActionDateTimeLong.lazySet(executeTime);
                }
                break;
            case ack:
                ackActionTableRowNumber.addAndGet(rowsCount);
                if (executeTime > 0)
                {
                    ackActionDateTimeLong.lazySet(executeTime);
                }
                break;
            default:
                break;
        }
    }

    // ================ setter / getter ==================
    public int getEventsSize()
    {
        return this.eventsSize;
    }

    public void setEventsSize(int eventsSize)
    {
        this.eventsSize = eventsSize;
    }

    public void setMemorySizePerEvent(int memorySizePerEvent)
    {
        this.memorySizePerEvent = memorySizePerEvent;
    }

    public void setDdlIsolation(boolean ddlIsolation)
    {
        this.ddlIsolation = ddlIsolation;
    }

    public AtomicLong getPutActionIndex()
    {
        return putActionIndex;
    }

    public AtomicLong getAckActionIndex()
    {
        return ackActionIndex;
    }

    public AtomicLong getPutActionMemorySize()
    {
        return putActionMemorySize;
    }

    public AtomicLong getAckActionMemorySize()
    {
        return ackActionMemorySize;
    }

    public StorageBatchMode getStorageBatchMode()
    {
        return storageBatchMode;
    }

    public void setStorageBatchMode(StorageBatchMode storageBatchMode)
    {
        this.storageBatchMode = storageBatchMode;
    }

    public AtomicLong getPutActionDateTimeLong()
    {
        return putActionDateTimeLong;
    }

    public AtomicLong getGetActionDateTimeLong()
    {
        return getActionDateTimeLong;
    }

    public AtomicLong getAckActionDateTimeLong()
    {
        return ackActionDateTimeLong;
    }

    public AtomicLong getPutActionTableRowNumber()
    {
        return putActionTableRowNumber;
    }

    public AtomicLong getGetActionTableRowNumber()
    {
        return getActionTableRowNumber;
    }

    public AtomicLong getAckActionTableRowNumber()
    {
        return ackActionTableRowNumber;
    }

}
